package com.deep.flink;


import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * https://www.dandelioncloud.cn/article/details/1616463118769291265
 */
public class HotPages {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置时间语义
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置并行度
        env.setParallelism(1);
        //读取文件 转换成POJO类型
        URL resource = HotPages.class.getResource("/apache.log");
        DataStream<String> inputStream = env.readTextFile(resource.getPath());
        DataStream<ApachLogEvent> dataStream = inputStream
                .map(line -> {
                    //分隔文件内容
                    String[] fields = line.split(" ");
                    //格式化时间格式
                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                    Long timestamp = simpleDateFormat.parse(fields[3]).getTime();
                    return new ApachLogEvent(fields[0], fields[1], timestamp, fields[5], fields[6]);
                })
                //Watermark 设置延迟时间
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ApachLogEvent>(Time.seconds(60)) {
                    @Override
                    public long extractTimestamp(ApachLogEvent apachLogEvent) {
                        return apachLogEvent.getTimestamp();
                    }
                });
        //分组开窗聚合
        //首先过滤 只要get请求即可
        SingleOutputStreamOperator<PageViewCount> windowAggStream = dataStream.filter(data -> "GET".equals(data.getMethod()))
                //按照url分组
                //.keyBy(data ->data.getUrl())
                .keyBy(ApachLogEvent::getUrl)
                //.keyBy("url")
                //开窗 10分钟统计一次 5秒钟滑动一次
                .timeWindow(Time.minutes(10), Time.seconds(5))
                .aggregate(new PageCountAgg(), new PageCountResult());
        //收集同一窗口count数据，排序输出
        DataStream<String> resultStream = windowAggStream.keyBy(PageViewCount::getWindowEnd)
                .process(new TopNHotPages(3));
        resultStream.print();
        env.execute("hot pages job");
    }

    //自定义预聚合函数
    public static class PageCountAgg implements AggregateFunction<ApachLogEvent, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(ApachLogEvent apachLogEvent, Long aLong) {
            return aLong + 1;
        }

        @Override
        public Long getResult(Long aLong) {
            return aLong;
        }

        @Override
        public Long merge(Long aLong, Long acc1) {
            return aLong + acc1;
        }
    }

    //实现自定义窗口函数
    public static class PageCountResult implements WindowFunction<Long, PageViewCount, String, TimeWindow> {
        @Override
        public void apply(String s, TimeWindow timeWindow, Iterable<Long> iterable, Collector<PageViewCount> collector) throws Exception {
            collector.collect(new PageViewCount(s, timeWindow.getEnd(), iterable.iterator().next()));
            ;
        }
    }

    //实现自定义的窗口函数
    public static class TopNHotPages extends KeyedProcessFunction<Long, PageViewCount, String> {
        private static Integer topSize;

        public TopNHotPages(int i) {
            topSize = i;
        }

        //定义状态 保存当前所有PageViewCount到List中
        ListState<PageViewCount> pageViewCountListState;

        @Override
        public void processElement(PageViewCount pageViewCount, Context context, Collector<String> collector) throws Exception {
            pageViewCountListState.add(pageViewCount);
            //定时器+1
            context.timerService().registerProcessingTimeTimer(pageViewCount.getWindowEnd() + 1);
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            pageViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<PageViewCount>("page-count-list", PageViewCount.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            ArrayList<PageViewCount> pageViewCounts = Lists.newArrayList(pageViewCountListState.get().iterator());
            //排序
            pageViewCounts.sort(new Comparator<PageViewCount>() {
                @Override
                public int compare(PageViewCount o1, PageViewCount o2) {
                    if (o1.getCount() > o2.getCount()) {
                        //不用排序
                        return -1;
                    } else if (o1.getCount() < o2.getCount()) {
                        return 1;
                    } else {
                        return 0;
                    }
                }
            });
            //格式化成字符串输出
            //排名信息格式化成string，方便打印输出
            StringBuilder resultBuilder = new StringBuilder();
            resultBuilder.append("==========");
            resultBuilder.append("窗口结束时间: ").append(new Timestamp(timestamp - 1)).append("\n");
            //遍历列表，取topN输出
            for (int i = 0; i < Math.min(topSize, pageViewCounts.size()); i++) {
                PageViewCount currentItemViewCount = pageViewCounts.get(i);
                resultBuilder.append("NO ").append(i + 1).append(":")
                        .append(" url=").append(currentItemViewCount.getUrl())
                        .append("热门度: ").append(currentItemViewCount.getCount())
                        .append("\n");
            }
            resultBuilder.append("==========\n");
            Thread.sleep(1000L);
            out.collect(resultBuilder.toString());
        }
    }
}